package com.hjy.backend.indexManager;


import com.hjy.backend.commons.SubArray;
import com.hjy.backend.dataManager.dataItem.DataItem;
import com.hjy.backend.transactionManager.TransactionManagerImpl;
import com.hjy.backend.utils.Parser;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * Node结构如下：
 * [LeafFlag][KeyNumber][SiblingUid]
 * [Son0][Key0][Son1][Key1]...[SonN][KeyN]
 * LeafFlag: 1位
 * KeyNumber: 2位
 * SiblingUid: 8位
 * [Son0][Key0]: 一个 Son 或 一个Key 占据 8位, SON 中存储uid, key是对应的编号, 用于快速找到 SON 的, 而且key有序
 * 叶子节点为 1, 非叶子节点为0
 */
public class Node {
    static final int IS_LEAF_OFFSET = 0;
    static final int NO_KEYS_OFFSET = IS_LEAF_OFFSET + 1;
    static final int SIBLING_OFFSET = NO_KEYS_OFFSET + 2;
    static final int NODE_HEADER_SIZE = SIBLING_OFFSET + 8;

    static final int BALANCE_NUMBER = 32;
    static final int NODE_SIZE = NODE_HEADER_SIZE + (2 * 8) * (BALANCE_NUMBER * 2 + 2); // 一个节点的大小, 能够存储的最大数据项为 64个

    /**
     * 对应的 B+ tree
     */
    BPlusTree tree;
    DataItem dataItem; // 对应的 dataItem
    SubArray raw;
    long uid; // uid 是 dataItem 的唯一标识

    /**
     * 节点也分叶子节点和非叶子节点, 只有叶子节点里面的的数据项可以存储数据
     *
     * @param raw
     * @param isLeaf
     */
    static void setRawIsLeaf(SubArray raw, boolean isLeaf) {
        if (isLeaf) {
            raw.raw[raw.start + IS_LEAF_OFFSET] = (byte) 1;
        } else {
            raw.raw[raw.start + IS_LEAF_OFFSET] = (byte) 0;
        }
    }

    /**
     * 判断是不是叶子节点
     *
     * @param raw
     * @return
     */
    static boolean getRawIsLeaf(SubArray raw) {
        return raw.raw[raw.start + IS_LEAF_OFFSET] == (byte) 1;
    }

    /**
     * 设置改节点的数据项数量
     *
     * @param raw
     * @param noKeys
     */
    static void setRawNoKeys(SubArray raw, int noKeys) {
        System.arraycopy(
                Parser.short2Byte((short) noKeys), 0,
                raw.raw, raw.start + NO_KEYS_OFFSET,
                2
        );
    }

    /**
     * 获取该节点的数据项个数
     *
     * @param raw
     * @return
     */
    static int getRawNoKeys(SubArray raw) {
        return (int) Parser.parseShort(Arrays.copyOfRange(raw.raw, raw.start + NO_KEYS_OFFSET, raw.start + NO_KEYS_OFFSET + 2));
    }

    /**
     * 设置兄弟节点的uid
     *
     * @param raw
     * @param sibling
     */
    static void setRawSibling(SubArray raw, long sibling) {
        System.arraycopy(Parser.long2Byte(sibling), 0, raw.raw, raw.start + SIBLING_OFFSET, 8);
    }

    /**
     * 获取兄弟节点的uid
     *
     * @param raw
     * @return
     */
    static long getRawSibling(SubArray raw) {
        return Parser.parseLong(Arrays.copyOfRange(raw.raw, raw.start + SIBLING_OFFSET, raw.start + SIBLING_OFFSET + 8));
    }

    /**
     * 设置节点内的第 kth 个子数据项
     *
     * @param raw
     * @param uid
     * @param kth
     */
    static void setRawKthSon(SubArray raw, long uid, int kth) {
        int offset = raw.start + NODE_HEADER_SIZE + kth * (8 * 2);
        System.arraycopy(Parser.long2Byte(uid), 0, raw.raw, offset, 8);
    }

    /**
     * 获取节点内的第 kth 个子数据项
     *
     * @param raw
     * @param kth
     * @return
     */
    static long getRawKthSon(SubArray raw, int kth) {
        int offset = raw.start + NODE_HEADER_SIZE + kth * (8 * 2);
        return Parser.parseLong(Arrays.copyOfRange(raw.raw, offset, offset + 8));
    }

    /**
     * 设置节点内的第 kth 个key
     *
     * @param raw
     * @param key
     * @param kth
     */
    static void setRawKthKey(SubArray raw, long key, int kth) {
        int offset = raw.start + NODE_HEADER_SIZE + kth * (8 * 2) + 8;
        System.arraycopy(Parser.long2Byte(key), 0, raw.raw, offset, 8);
    }

    /**
     * 获取节点内的第 kth 个key
     *
     * @param raw
     * @param kth
     * @return
     */
    static long getRawKthKey(SubArray raw, int kth) {
        int offset = raw.start + NODE_HEADER_SIZE + kth * (8 * 2) + 8;
        return Parser.parseLong(Arrays.copyOfRange(raw.raw, offset, offset + 8));
    }

    /**
     * 将第kth个后的数据项复制到新的SubArray
     *
     * @param from
     * @param to
     * @param kth
     */
    static void copyRawFromKth(SubArray from, SubArray to, int kth) {
        int offset = from.start + NODE_HEADER_SIZE + kth * (8 * 2);
        System.arraycopy(from.raw, offset, to.raw, to.start + NODE_HEADER_SIZE, from.end - offset);
    }

    /**
     * 从kth到后, 所有数据项向后移动 16 位, 给第 kth 个数据项留出空间
     *
     * @param raw
     * @param kth
     */
    static void shiftRawKth(SubArray raw, int kth) {
        int begin = raw.start + NODE_HEADER_SIZE + (kth + 1) * (8 * 2);
        int end = raw.start + NODE_SIZE - 1;
        for (int i = end; i >= begin; i--) {
            raw.raw[i] = raw.raw[i - (8 * 2)];
        }
    }

    /**
     * 生成一个更新后的根节点的数据, 第一个uid是根uid, 第二个uid是最后一个节点的uid
     * 新节点产生的 key 放到第一个数据项中, 有助于搜索和插入时找到对应的节点(算法决定的)
     * @param left 第一个uid
     * @param right 第二个uid
     * @param key 第一个uid对应的key
     * @return
     */
    static byte[] newRootRaw(long left, long right, long key) {
        SubArray raw = new SubArray(new byte[NODE_SIZE], 0, NODE_SIZE);

        setRawIsLeaf(raw, false);
        setRawNoKeys(raw, 2); // 目前节点里面有两个数据项(key)
        setRawSibling(raw, 0);
        setRawKthSon(raw, left, 0);
        setRawKthKey(raw, key, 0);
        setRawKthSon(raw, right, 1);
        // 所有的节点的 key 都在 0 ~ Long.MAX_VALUE , 可以找到任何的索引
        setRawKthKey(raw, Long.MAX_VALUE, 1);

        return raw.raw;
    }

    /**
     * 生成一个空的根节点数据
     *
     * @return
     */
    static byte[] newNilRootRaw() {
        SubArray raw = new SubArray(new byte[NODE_SIZE], 0, NODE_SIZE);

        setRawIsLeaf(raw, true);
        setRawNoKeys(raw, 0);
        setRawSibling(raw, 0);

        return raw.raw;
    }

    /**
     * 通过 uid 加载出一个 新的Node
     *
     * @param bTree
     * @param uid
     * @return
     * @throws Exception
     */
    static Node loadNode(BPlusTree bTree, long uid) throws Exception {
        DataItem di = bTree.dm.read(uid);
        assert di != null;
        Node n = new Node();
        n.tree = bTree;
        n.dataItem = di;
        n.raw = di.data();
        n.uid = uid;
        return n;
    }

    /**
     * 释放掉对应的 dataItem
     */
    public void release() {
        dataItem.release();
    }

    /**
     * 是否是叶子节点
     *
     * @return
     */
    public boolean isLeaf() {
        dataItem.rLock();
        try {
            return getRawIsLeaf(raw);
        } finally {
            dataItem.rUnLock();
        }
    }

    class SearchNextRes {
        long uid;
        long siblingUid;
    }

    /**
     * searchNext 寻找对应 key(数据项) 的 UID, 如果找不到, 则返回兄弟节点的 UID
     *
     * @param key
     * @return
     */
    public SearchNextRes searchNext(long key) {
        dataItem.rLock();
        try {
            SearchNextRes res = new SearchNextRes();
            int noKeys = getRawNoKeys(raw);
            for (int i = 0; i < noKeys; i++) {
                long ik = getRawKthKey(raw, i);
                // 找到了下一个数据项
                if (key < ik) {
                    res.uid = getRawKthSon(raw, i);
                    res.siblingUid = 0;
                    return res;
                }
            }
            res.uid = 0;
            res.siblingUid = getRawSibling(raw);
            return res;

        } finally {
            dataItem.rUnLock();
        }
    }

    class LeafSearchRangeRes {
        List<Long> uids;
        long siblingUid;
    }

    /**
     * leafSearchRange 方法在当前节点进行范围查找，范围是 [leftKey, rightKey]，
     * 这里约定如果 rightKey 大于等于该节点的最大的 key, 则还同时返回兄弟节点的 UID，方便继续搜索下一个节点。
     *
     * @param leftKey
     * @param rightKey
     * @return
     */
    public LeafSearchRangeRes leafSearchRange(long leftKey, long rightKey) {
        dataItem.rLock();
        try {
            // 获取节点的数量
            int noKeys = getRawNoKeys(raw);
            int kth = 0;
            // 偏移到 leftKey 的位置
            while (kth < noKeys) {
                long ik = getRawKthKey(raw, kth);
                if (ik >= leftKey) {
                    break;
                }
                kth++;
            }
            List<Long> uids = new ArrayList<>();
            // 将 leftKey 和 rightKey 之间的所有 uid 存储起来
            while (kth < noKeys) {
                long ik = getRawKthKey(raw, kth);
                if (ik <= rightKey) {
                    uids.add(getRawKthSon(raw, kth));
                    kth++;
                } else {
                    break;
                }
            }
            long siblingUid = 0;
            // 好像找到节点末尾了, 就把下一个节点的uid返回, 方便继续查找
            if (kth == noKeys) {
                siblingUid = getRawSibling(raw);
            }
            LeafSearchRangeRes res = new LeafSearchRangeRes();
            res.uids = uids;
            res.siblingUid = siblingUid;
            return res;
        } finally {
            dataItem.rUnLock();
        }
    }

    /**
     * 如果被分割 newSon(新节点UID) 和 newKey 有内容, 如果这个节点不能插入, 那么siblingUid有内容
     */
    class InsertAndSplitRes {
        long siblingUid, newSon, newKey;
    }

    /**
     * 我们要向节点里面插入一个数据项key, 如果满了就新增一个Node
     *
     * @param uid
     * @param key
     * @return
     * @throws Exception
     */
    public InsertAndSplitRes insertAndSplit(long uid, long key) throws Exception {
        boolean success = false;
        Exception err = null;
        InsertAndSplitRes res = new InsertAndSplitRes();

        // 这个Node 被修改了, 对应的dataItem 需要记录下来
        dataItem.before();
        try {
            success = insert(uid, key);
            // 如果插入失败, 说明这个节点大小不够了, 返回兄弟节点继续尝试
            if (!success) {
                res.siblingUid = getRawSibling(raw);
                return res;
            }
            // 当前节点存储的数据项达到了上限, 需要分割出一个新的节点
            if (needSplit()) {
                try {
                    SplitRes r = split();
                    res.newSon = r.newSon;
                    res.newKey = r.newKey;
                    return res;
                } catch (Exception e) {
                    err = e;
                    throw e;
                }
            } else {
                return res;
            }
        } finally {
            if (err == null && success) {
                dataItem.after(TransactionManagerImpl.SUPER_XID);
            } else {
                dataItem.unBefore();
            }
        }
    }

    /**
     * 向节点中插入一个数据项
     *
     * @param uid
     * @param key
     * @return
     */
    private boolean insert(long uid, long key) {
        // 当前数据项个数
        int noKeys = getRawNoKeys(raw);
        int kth = 0;
        // 找到一个合适的地方插入(因为key需要有序)
        while (kth < noKeys) {
            long ik = getRawKthKey(raw, kth);
            if (ik < key) {
                kth++;
            } else {
                break;
            }
        }
        // 没找到好位置, 寄辣, 插入失败
        if (kth == noKeys && getRawSibling(raw) != 0) return false;

        // 开始插入
        if (getRawIsLeaf(raw)) {
            // 叶子节点, 直接梭哈
            shiftRawKth(raw, kth);
            setRawKthKey(raw, key, kth);
            setRawKthSon(raw, uid, kth);
            setRawNoKeys(raw, noKeys + 1);
        } else {
            // 不是叶子节点, kEY:原UID 原KEY:UID 很怪的实现, 但是没什么影响, key仍然有序
            long kk = getRawKthKey(raw, kth);
            setRawKthKey(raw, key, kth);
            shiftRawKth(raw, kth + 1);
            setRawKthKey(raw, kk, kth + 1);
            setRawKthSon(raw, uid, kth + 1);
            setRawNoKeys(raw, noKeys + 1);
        }
        return true;
    }

    /**
     * 当前节点存储的节点个数达到上限了64个
     *
     * @return
     */
    private boolean needSplit() {
        return BALANCE_NUMBER * 2 == getRawNoKeys(raw);
    }

    class SplitRes {
        long newSon, newKey; // 新节点的UID, 新节点第一个key
    }

    /**
     * 切割节点数据项, 每个节点分配 32 个
     *
     * @return
     * @throws Exception
     */
    private SplitRes split() throws Exception {
        // 新的共享数组
        SubArray nodeRaw = new SubArray(new byte[NODE_SIZE], 0, NODE_SIZE);
        // 如果当前节点为叶子节点, 就设置为叶子节点
        setRawIsLeaf(nodeRaw, getRawIsLeaf(raw));
        // 平分数据项, 每个节点32个
        setRawNoKeys(nodeRaw, BALANCE_NUMBER);
        // 设置兄弟节点, 注意!!!
        setRawSibling(nodeRaw, getRawSibling(raw));
        // 32~64 号数据项给新节点
        copyRawFromKth(raw, nodeRaw, BALANCE_NUMBER);
        // 创建新节点成功, 返回新节点对应的UID
        long newNodeUid = tree.dm.insert(TransactionManagerImpl.SUPER_XID, nodeRaw.raw);
        setRawNoKeys(raw, BALANCE_NUMBER);
        setRawSibling(raw, newNodeUid);

        SplitRes res = new SplitRes();
        res.newSon = newNodeUid;
        res.newKey = getRawKthKey(nodeRaw, 0);
        return res;
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append("Is leaf: ").append(getRawIsLeaf(raw)).append("\n");
        int KeyNumber = getRawNoKeys(raw);
        sb.append("KeyNumber: ").append(KeyNumber).append("\n");
        sb.append("sibling: ").append(getRawSibling(raw)).append("\n");
        for (int i = 0; i < KeyNumber; i++) {
            sb.append("son: ").append(getRawKthSon(raw, i)).append(", key: ").append(getRawKthKey(raw, i)).append("\n");
        }
        return sb.toString();
    }

}
